package com.l.im.service.handler;


import com.alibaba.fastjson.JSON;
import com.l.im.common.api.ApiResponse;
import com.l.im.common.cache.RedisService;
import com.l.im.common.config.ConfigReader;
import com.l.im.common.mq.RocketMQSender;
import com.l.im.model.SingleMessageDTO;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: LIM
 * @description: 消息分发
 * @author: Alex Wu
 * @createDate: 2025-04-16 11:55
 **/
@Slf4j
public class MessageDispatcher {

    private static final Map<String, Channel> userChannelMap = new ConcurrentHashMap<>();
    private static final Map<Channel, String> channelUsers = new ConcurrentHashMap<>();

    private static final RedisService redisService = new RedisService();

    public static void register(String userId, Channel channel) {
        userChannelMap.put(userId, channel);
        channelUsers.put(channel, userId);
        log.info("用户 {} 注册连接", userId);
    }

    public static void unregister(Channel channel) {
        String userId = channelUsers.remove(channel);
        if (userId != null) {
            userChannelMap.remove(userId);
            log.info("用户 {} 断开连接", userId);
        }
    }

    public static boolean isUserOnline(String userId) {
        return userChannelMap.containsKey(userId);
    }

    public static void sendToUser(String userId, ApiResponse<?> response) {
        Channel channel = userChannelMap.get(userId);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(new TextWebSocketFrame(response.toString()));
            log.info("向用户 {} 发送消息：{}", userId, response);
        } else {
            log.info("用户 {} 不在线，保存离线消息", userId);
            saveOfflineMessage(userId, response);
        }
    }

    public static void saveOfflineMessage(String userId, ApiResponse<?> response) {
        String key = "offline:msg:" + userId;
        String messageJson = JSON.toJSONString(response);
        redisService.lpush(key, messageJson);  // 将消息压入队列
        redisService.expire(key, 3600L);        // 设置1小时过期（可根据业务调整）
        log.info("已保存离线消息：{}", messageJson);
    }

    public static List<String> getOfflineMessages(String userId) {
        String key = "offline:msg:" + userId;
        List<String> messages = redisService.lrange(key, 0, -1);
        redisService.del(key); // 拉取后删除
        log.info("拉取到 {} 条离线消息", messages.size());
        return messages;
    }

    public static void dispatchToUser(String fromUserId, String toUserId, String message) throws ClientException {
        // 如果本节点连接存在，直接投送
        if (isUserOnline(toUserId)) {
            sendToUser(toUserId, ApiResponse.success(new SingleMessageDTO("即时消息", fromUserId, toUserId, message)));
            return;
        }

        // 跨服或离线处理逻辑
        String userMeta = redisService.get("online:user:" + toUserId);
        if (userMeta != null) {
            String[] parts = userMeta.split(":");
            if (parts.length >= 2) {
                String topic = parts[1];
                RocketMQSender sender = RocketMQSender.getInstance(
                        ConfigReader.getInstance().getOrDefault("rocketmq.endpoint", String.class, ""),
                        topic
                );
                String mqKey = fromUserId + ":" + toUserId;
                String mqMessage = JSON.toJSONString(new SingleMessageDTO("跨服投递", fromUserId, toUserId, message));
                sender.send("imTag", mqKey, mqMessage);
                log.info("用户 {} 在其他节点，消息已通过 MQ 转发", toUserId);
                return;
            }
        }
        // 用户确实不在线，存入离线消息
        log.info("用户 {} 完全不在线，消息转离线存储", toUserId);
        saveOfflineMessage(toUserId, ApiResponse.success(new SingleMessageDTO("离线消息", fromUserId, toUserId, message)));
    }

}
